package com.itlong.cloud.iot.alibaba.mns;

import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.model.Message;
import org.slf4j.Logger;import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * <desc>
 *      监听阿里云mns消息队列多线程处理类
 * </desc>
 *
 * @createDate 2018/04/09
 */
public class CloudQueueMessageReceiver {
    private static final Logger LOG = LoggerFactory.getLogger(CloudQueueMessageReceiver.class);
    public static final int WAIT_SECONDS = 30;              //等待消息时长

    protected static final Map<String, Object> sLockObjMap = new HashMap<String, Object>();        //是否有对象占用轮询锁
    protected static Map<String, Boolean> sPollingMap = new ConcurrentHashMap<String, Boolean>();    //是否有线程正在轮询

    protected Object lockObj;    //锁
    protected String queueName; //队列名称
    protected CloudQueue cloudQueue; //队列
    protected int workerId; //线程标识

    /**
     * <desc>
     *      初始化
     * </desc>
     *
     * @param id 线程id
     * @param mnsClient 消息队列消费者客户端
     * @param queue 队列名称
     * @author Jiaqi.X
     * @createDate 2018/04/09
     */
    public CloudQueueMessageReceiver(int id, MNSClient mnsClient, String queue) {
        if (queue == null || queue.equals("")) {
            return;
        }
        cloudQueue = mnsClient.getQueueRef(queue);
        queueName = queue;
        workerId = id;
        //判断当前是否有线程开始轮询消息
        synchronized (sLockObjMap) {
            lockObj = sLockObjMap.get(queueName);
            if (lockObj == null) {
                lockObj = new Object();
                sLockObjMap.put(queueName, lockObj);
            }
        }
    }


    /**
     * <desc>
     *      判断线程当前轮询状态，如果队列没有消息，则只让一个线程进行消息的轮询，否则所有线程唤醒即时接收队列消息
     * </desc>
     *
     * @return
     * @author Jiaqi.X
     * @createDate 2018/04/10
     */
    public boolean setPolling() {
        synchronized (lockObj) {
            Boolean ret = sPollingMap.get(queueName);
            if (ret == null || !ret) {
                sPollingMap.put(queueName, true);
                return true;
            }
            return false;
        }
    }

    /**
     * <desc>
     *      监听到队列有消息后，唤醒所有线程
     * </desc>
     *
     * @author Jiaqi.X
     * @createDate 2018/04/10
     */
    public void clearPolling() {
        synchronized (lockObj) {
            sPollingMap.put(queueName, false);
            lockObj.notifyAll();
//            LOG.info(String.format("【IOT MNS消息读取唤醒所有线程】"));
        }
    }

    /**
     * <desc>
     *      从队列中读取消息
     * </desc>
     *
     * @return
     * @author Jiaqi.X
     * @createDate 2018/04/10
     */
    public Message receiveMessage() {
        //初始化轮询状态
        boolean polling = false;

        while (true) {
            //线程同步执行判断是否要进入等待挂起状态
            synchronized (lockObj) {
                Boolean p = sPollingMap.get(queueName);
                //判断当前是否有线程正在轮询等待消息队列消息
                if (p != null && p) {
                    try {
//                        LOG.info(String.format("【IOT MNS消息读取Thread%s进入等待状态】", workerId));
                        polling = false;
                        //线程进入等待状态，同时释放锁
                        lockObj.wait();
                    } catch (InterruptedException e) {
                        LOG.error(String.format("【IOT MNS消息读取中断，队列名为%s】", queueName));
                        return null;
                    }
                }
            }
            try {
                Message message = null;
                if (!polling) {
                    //如果当前消息队列有数据，线程直接获取消息队列中的数据，不使用长连接监听队列
                    message = cloudQueue.popMessage();
                    //如果消息队列返回消息为null，设置当前轮询状态为true，确保下次线程获取消息时只启动一个线程进行消息的监听
                    if (message == null) {
                        polling = true;
                        continue;
                    }
                } else {
                    //设置当前线程为长连接监听队列消息状态 此方法为同步方法
                    if (setPolling()) {
//                        LOG.info(String.format("【IOT MNS消息读取单线程监听中，线程id为%s】", workerId));
                    } else {
                        continue;
                    }
                    do {
//                        LOG.info(String.format("【IOT MNS消息读取单线程保持监听中，线程id为%s】", workerId));
                        try {
                            message = cloudQueue.popMessage(WAIT_SECONDS);
                        } catch(Exception e) {
                            LOG.error(String.format("【IOT MNS消息读取单线程监听异常，线程id为%s】", workerId));
                        }
                    } while (message == null);
                    //当监听到消息后唤醒所有线程进入消息读取竞争状态，直到没有消息再将其中某一线程保持长连接轮询状态，其他线程全部等待挂起
                    clearPolling();
                }
                return message;
            } catch (Exception e) {
                LOG.error(String.format("【IOT MNS消息读取网络异常，线程id为%s】", workerId),e);
            }
        }
    }

}
